package com.yb.kafka.consumer.listener;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Duration;
import java.util.Objects;
import java.util.Optional;

/**
  * @Description: 消费者---使用@KafkaListener注解,可以指定:主题,分区,消费组
  * @author yangbiao
  * @date 2018/11/6
 */
@Slf4j
@Service
public class KafkaTwoConsumer {

    /**
     * 这个简单点,只能获取到消息值
     * @param message
     */
    @KafkaListener(topics = {"myMessage"},groupId = "cc")
    public void receive(String message){
        JSONObject jsonObject = JSONObject.parseObject(message);
        System.err.println("消费者Two消费消息:" + jsonObject);
    }

    /**
     * ======================需要开启手动ack也就是配置spring.kafka.listener.ack-mode=manual
     * 这个可以获取到除了消息的其他东西--->上面消费完了,下面就没得消费了,因为主题是一样的
     * @param record
     */
    @Transactional(rollbackFor = Exception.class)
    @KafkaListener(topics = {"myMessage"},groupId = "cc")
    public void consumerMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        //默认设置可以手动提交offset
        boolean flag = true;
        try {
            if(Objects.nonNull(record)){
                log.info("消费Two的key:" + record.key());
                log.info("消费Two的value:" + record.value());
            }else {
                log.info("获取到需要消费的消息为空");
            }
            //throw new IllegalArgumentException("");
        } catch (Exception e) {
            //消费业务逻辑出现异常,不手动提交offset
            flag=false;
            log.info("consumer-two消费异常"+e.getMessage());
            //throw new IllegalArgumentException("我是consumer-two的异常");
        } finally {
            //处理offset的提交
            if (flag){
                //手动提交offset(调用此方法意味着已经处理了分区中以前的所有消息)----->主动告知消息已经消费
                ack.acknowledge();
            }
        }
    }
}